package com.atguigu.canal.clients;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.atguigu.realtime.constants.PrefixConstant;
import com.atguigu.realtime.constants.TopicConstant;
import com.atguigu.realtime.utils.KafkaProducerUtil;
import com.atguigu.realtime.utils.RedisUtil;
import com.google.protobuf.ByteString;
import com.google.protobuf.InvalidProtocolBufferException;
import redis.clients.jedis.Jedis;

import java.net.InetSocketAddress;
import java.util.List;
import java.util.Random;

/**
 * Created by Smexy on 2022/7/19
 *
 *  订阅 order_info和order_detail 的 insert操作 ,将信息发送到kafka
 *  订阅 user_info的insert和update操作，将信息发送到redis
 *          在redis怎么存？
 *              当前的需求是，根据userId查询用户信息
 *  			key:  userinfo:userId
 * 			    value:  string
 */
public class SaleDetailClient {

    private static Jedis jedis = RedisUtil.getJedis();

    public static void main(String[] args) throws InterruptedException, InvalidProtocolBufferException {

        CanalConnector canalConnector = CanalConnectors.newSingleConnector(new InetSocketAddress("hadoop103", 11111),
                "example", null
                , null
        );


        canalConnector.connect();

        //③订阅表  格式:库名.表名
        canalConnector.subscribe("220309.*");

        //④拉取数据，拉到解析订阅到的数据
        while (true){

            Message message = canalConnector.get(100);

            if (message.getId() == -1){
                //没拉到数据  单押
                System.out.println("当前没有新数据，歇5秒再去");
                //歇会
                Thread.sleep(5000);

                //跳过本次循环，开始下次循环
                continue;
            }


            //打印数据
            //System.out.println(message);
            List<CanalEntry.Entry> entries = message.getEntries();

            for (CanalEntry.Entry entry : entries) {

                String tableName = entry.getHeader().getTableName();

                //是rowdata，才可能是 insert | update | delete
                if (entry.getEntryType().equals(CanalEntry.EntryType.ROWDATA)){

                    parseData(entry.getStoreValue(),tableName);
                }

            }


        }

    }


    private static void parseData(ByteString storeValue,String tableName) throws InvalidProtocolBufferException {

        CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(storeValue);

        if ("order_info".equals(tableName) && rowChange.getEventType().equals(CanalEntry.EventType.INSERT)){

            sendData(rowChange,TopicConstant.ORDER_INFO,true);

        }else  if ("order_detail".equals(tableName) && rowChange.getEventType().equals(CanalEntry.EventType.INSERT)){

            sendData(rowChange,TopicConstant.ORDER_DETAIL,true);

        }else  if ("user_info".equals(tableName) &&
                (rowChange.getEventType().equals(CanalEntry.EventType.INSERT) ||rowChange.getEventType().equals(CanalEntry.EventType.UPDATE)  )){

            sendData(rowChange,null,false);
        }



    }

    private static void sendData(CanalEntry.RowChange rowChange,String topic,boolean ifSendToKafka){

        //多行数据变化
        List<CanalEntry.RowData> rowDatasList = rowChange.getRowDatasList();

        // rowData: 1行数据变化
        for (CanalEntry.RowData rowData : rowDatasList) {

            //构建一个JSONObject 希望把每一行封装为一个{}
            JSONObject jsonObject = new JSONObject();

            List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();

            for (CanalEntry.Column column : afterColumnsList) {

                jsonObject.put(column.getName(), column.getValue());

            }

            if (ifSendToKafka){

                //模拟随机的延迟
                int i = new Random().nextInt(5);

                /*try {
                    Thread.sleep(i * 1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }*/

                KafkaProducerUtil.sendData(jsonObject.toString(), topic);
            }else{

                jedis.set(PrefixConstant.user_info_redis_preffix + jsonObject.getString("id"),jsonObject.toString());
            }


        }


    }
}
